package com.zdxk.msgPush.common.load;

import com.zdxk.msgPush.common.config.RocketMqConfig;
import com.zdxk.msgPush.common.rocket.consume.RocketMqPushConsumer;
import com.zdxk.msgPush.common.thread.ExecutorProcessPool;
import com.zdxk.msgPush.msg.subject.StockTradeMsgTask;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;


@Component
public class RocketMqComsumerLoad implements ApplicationRunner {

    private static Logger logger = LoggerFactory.getLogger(RocketMqComsumerLoad.class);

    @Override
    public void run(ApplicationArguments args) throws Exception {
        ExecutorProcessPool processPool = ExecutorProcessPool.getInstance();
        String groupName = RocketMqConfig.getConf("rocketMq.producer.groupName");
        String nameSvrAddr = RocketMqConfig.getConf("rocketMq.nameServer.domain");
        RocketMqPushConsumer consumer = new RocketMqPushConsumer(groupName, nameSvrAddr, (list, consumeConcurrentlyContext) -> {
            for(MessageExt obj : list){
                String msg = new String(obj.getBody());
                logger.debug(msg);
                processPool.execute(new StockTradeMsgTask(msg));
            }
        });
        consumer.subscribe(RocketMqConfig.getConf("rocketMq.topic"),"*");
        consumer.start();
    }
}
